package HandshakeJoin;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.lang.StringUtils;
import com.google.common.collect.Lists;
import com.google.common.primitives.Ints;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.logging.impl.Log4JLogger;
import org.apache.log4j.Level;

public class HBaseClient implements DBClient
{
    transient private static final Logger logger = LoggerFactory
            .getLogger(HBaseClient.class);

    transient private static final int TIMESTAMP_STRING_SZ = 19;

    private static Configuration conf;

    public static void initGlobal() {
        setHadoopLogLevel(Level.INFO);

        conf = HBaseConfiguration.create();
    }

    private static void setHadoopLogLevel(Level level) {
        Log hadoopLog = LogFactory.getLog("org");
        ((Log4JLogger) hadoopLog).getLogger().setLevel(level);

        logger.info("Hadoop log level: " + level);
    }

    private HTable table;
    private long lastTimestamp = -1;
    private int lastSeq;

    public HBaseClient(final String tableName) {
        try {
            table = new HTable(conf, tableName);
        }
        catch (IOException e) {
            e.printStackTrace();
        }
    }

    public boolean init() {
        // delete existing table with same name, if any
        try {
            deleteTable();
        }
        catch (Exception e) {
            logger.warn(e.toString());
        }

        // create a new table
        boolean rt = false;
        try {
            String[] familys = { "main", "auxiliary" };

            rt = createTable(familys);
        }
        catch (Exception e) {
            e.printStackTrace();
        }

        return rt;
    }

    public void close() {}

    public boolean materialize(Snapshot snapshot) {
        try {
            insertRecords(snapshot.getRecords(), snapshot.getTimeStamp());
        }
        catch (Exception e) {
            e.printStackTrace();
        }

        return true;
    }

    public List<ArrayList<Integer>> retrieveByTimestamp(long ts) {
        try {
            return selectByTimestamp(ts, ts);
        }
        catch (Exception e) {
            return null;
        }
    }

    public List<ArrayList<Integer>> retrieveByTimestamp(long tsLower,
            long tsUpper) {
        try {
            return selectByTimestamp(tsLower, tsUpper);
        }
        catch (Exception e) {
            return null;
        }
    }

    public List<ArrayList<Integer>> retrieveByJoinKey(int val) {
        try {
            return selectByJoinKey(val, val);
        }
        catch (Exception e) {
            return null;
        }
    }

    public List<ArrayList<Integer>> retrieveByJoinKey(int valLower, int valUpper) {
        try {
            return selectByJoinKey(valLower, valUpper);
        }
        catch (Exception e) {
            return null;
        }
    }

    public boolean createTable(String[] familys) throws Exception {
        boolean rt = false;
        HBaseAdmin admin = new HBaseAdmin(conf);
        byte[] tableName = table.getTableName();

        if (admin.tableExists(tableName)) {
            logger.warn("table already exists!");
            rt = false;
        }
        else {
            HTableDescriptor tableDesc = new HTableDescriptor(tableName);

            for (int i = 0; i < familys.length; ++i) {
                tableDesc.addFamily(new HColumnDescriptor(familys[i]));
            }
            admin.createTable(tableDesc);
            rt = true;
        }

        if (admin.isTableDisabled(tableName)) {
            admin.enableTable(tableName);
        }

        return rt;
    }

    public void deleteTable() throws Exception {
        HBaseAdmin admin = new HBaseAdmin(conf);
        byte[] tableName = table.getTableName();

        if (admin.tableExists(tableName)) {
            if (admin.isTableEnabled(tableName)) {
                admin.disableTable(tableName);
            }

            admin.deleteTable(tableName);
        }
    }

    private void insertRecords(final List<ArrayList<Integer>> records,
            final long timestamp) throws Exception {
        int seq;
        if (timestamp == lastTimestamp) {
            seq = lastSeq;
        }
        else {
            seq = 0;
        }

        String tsString = StringUtils.leftPad(Long.toString(timestamp),
                TIMESTAMP_STRING_SZ, '0');
        byte[] valueBytes;

        for (ArrayList<Integer> record : records) {
            ++seq;
            byte[] rowBytes = Bytes.toBytes(tsString + "_" + seq);

            for (int i = 0; i < record.size(); ++i) {
                valueBytes = Bytes.toBytes(record.get(i));
                insert(rowBytes, "main", "col_" + i, valueBytes);
            }

            // Note: No need to specially store the timestamp since we already
            // use it to identify the row key.
            // valueBytes = Bytes.toBytes(timestamp);
            // insert(rowBytes, "auxiliary", "timestamp", valueBytes);
        }

        lastTimestamp = timestamp;
        lastSeq = seq;
    }

    private void insert(final byte[] rowBytes, final String family,
            final String qualifier, final byte[] valueBytes) throws IOException {
        byte[] familyBytes = Bytes.toBytes(family);
        byte[] qualifierBytes = Bytes.toBytes(qualifier);

        Put put = new Put(rowBytes);
        put.add(familyBytes, qualifierBytes, valueBytes);

        table.put(put);
    }

    private List<ArrayList<Integer>> selectByTimestamp(final long tsLower,
            final long tsUpper) throws IOException {
        List<ArrayList<Integer>> records = Lists.newLinkedList();
        String tsLowerString = StringUtils.leftPad(Long.toString(tsLower),
                TIMESTAMP_STRING_SZ, '0');
        Scan s = new Scan(Bytes.toBytes(tsLowerString));

        s.addFamily(Bytes.toBytes("main"));
        ResultScanner rsScanner = table.getScanner(s);

        for (Result rs : rsScanner) {
            String row = Bytes.toString(rs.getRow());
            long ts = Long.parseLong(row.split("_")[0]);

            if (ts >= tsLower && ts <= tsUpper) {
                records.add(getRecord(rs));
            }
            else {
                break;
            }
        }

        return records;
    }

    @Deprecated
    /**
     * This method is of low efficiency because it fetches records one-by-one 
     * through iterative random access.
     */
    private List<ArrayList<Integer>> selectByTimestamp(final long timestamp)
            throws Exception {
        List<ArrayList<Integer>> records = Lists.newLinkedList();
        int seq = 0;
        ArrayList<Integer> record = getOneRecordByTimestamp(timestamp, seq);

        while (record != null) {
            records.add(record);
            record = getOneRecordByTimestamp(timestamp, ++seq);
        }

        return records;
    }

    private ArrayList<Integer> getOneRecordByTimestamp(final long timestamp,
            final int seq) throws Exception {
        String tsString = Long.toString(timestamp);
        byte[] rowBytes = Bytes.toBytes(tsString + "_" + seq);
        Get get = new Get(rowBytes);
        Result rs = table.get(get);

        ArrayList<Integer> record;
        if (rs.isEmpty()) {
            return null;
        }
        else {
            record = Lists.newArrayListWithCapacity(rs.size());
        }

        for (KeyValue kv : rs.raw()) {
            record.add(Bytes.toInt(kv.getValue()));
        }

        return record;
    }

    private List<ArrayList<Integer>> selectByJoinKey(final int valLower,
            final int valUpper) throws IOException {
        List<ArrayList<Integer>> records = Lists.newLinkedList();
        byte[] familyBytes = Bytes.toBytes("main");
        byte[] qualifierBytes = Bytes.toBytes("col_0");
        ResultScanner rsScanner = table.getScanner(familyBytes);

        for (Result rs : rsScanner) {
            byte[] valueBytes = rs.getValue(familyBytes, qualifierBytes);
            int value = Bytes.toInt(valueBytes);

            if (value >= valLower && value <= valUpper) {
                records.add(getRecord(rs));
            }
        }

        return records;
    }

    private ArrayList<Integer> getRecord(final Result rs) {
        ArrayList<Integer> record = Lists.newArrayListWithCapacity(rs.size());

        for (KeyValue kv : rs.raw()) {
            record.add(Bytes.toInt(kv.getValue()));
        }

        return record;
    }

    /**
     * Test client.
     */
    public static void main(String[] args) throws Exception {
        HBaseClient.initGlobal();

        DBClient db = new HBaseClient("test");
        db.init();

        int[][] data = { { 1, 2, 3 }, { 1, 3, 4 }, { 2, 5, 3 }, { 3, 2, 6 },
                { 1, 3, 2 }, { 4, 5, 1 }, { 3, 7, 2 }, { 5, 4, 3 } };
        List<ArrayList<Integer>> records = Lists.newLinkedList();
        for (int i = 0; i < data.length; ++i) {
            records.add(Lists.newArrayList(Ints.asList(data[i])));
        }
        Snapshot snapshot = new Snapshot("test", records);

        db.materialize(snapshot);

        System.out.println("@@@ Test: Equal selection wrt Join key @@@");
        records = db.retrieveByJoinKey(1);
        for (ArrayList<Integer> record : records) {
            System.out.println(record);
        }

        System.out.println("@@@ Test: Range selection wrt Join key @@@");
        records = db.retrieveByJoinKey(2, 3);
        for (ArrayList<Integer> record : records) {
            System.out.println(record);
        }

        System.out.println("@@@ Test: Equal selection wrt Timestamp @@@");
        records = db.retrieveByTimestamp(snapshot.getTimeStamp());
        for (ArrayList<Integer> record : records) {
            System.out.println(record);
        }

        System.out.println("@@@ Test: Range selection wrt Timestamp @@@");
        snapshot.setTimestamp(1);
        db.materialize(snapshot);

        snapshot.setTimestamp(2);
        db.materialize(snapshot);

        snapshot.setTimestamp(10);
        db.materialize(snapshot);

        snapshot.setTimestamp(11);
        db.materialize(snapshot);

        records = db.retrieveByTimestamp(3, 12);
        for (ArrayList<Integer> record : records) {
            System.out.println(record);
        }

        db.close();
    }
}